package com.example;

import com.example.utils.ConnectionFactoryUtils;
import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.UUID;
import java.util.concurrent.*;

/**
 * @author wangxiyuan
 * @description
 * @createDate 2021/3/24 19:30
 **/
public class RPCClient implements AutoCloseable{
    private Connection connection;
    private Channel channel;
    private String requestQueueName = "rpc_queue";

    public RPCClient() throws IOException, TimeoutException, InterruptedException {
        ConnectionFactory factory = ConnectionFactoryUtils.getConnectionFactory();

        connection = factory.newConnection();
        channel = connection.createChannel();
    }

    public static void main(String[] argv) {
        try (RPCClient fibonacciRpc = new RPCClient()) {
            for (int i = 0; i < 32; i++) {
                String i_str = Integer.toString(i);
                System.out.println(" [x] Requesting fib(" + i_str + ")");
                String response = fibonacciRpc.call(i_str);
                System.out.println(" [.] Got '" + response + "'");
            }
        } catch (IOException | TimeoutException | InterruptedException e) {
            e.printStackTrace();
        }
    }

    public String call(String message) throws IOException, InterruptedException {
        final String corrId = UUID.randomUUID().toString();

        String replyQueueName = "amq.rabbitmq.reply-to";//channel.queueDeclare().getQueue();
        AMQP.BasicProperties props = new AMQP.BasicProperties
                .Builder()
                .correlationId(corrId)
                .replyTo(replyQueueName)
                .build();
        final BlockingQueue<String> response = new ArrayBlockingQueue<>(1);
        // basicConsume需要放在basicPublish的前面，不然会报错
        String ctag = channel.basicConsume(replyQueueName, true, (consumerTag, delivery) -> {
            try {
                if (delivery.getProperties().getCorrelationId().equals(corrId)) {
                    response.offer(new String(delivery.getBody(), "UTF-8"));
                }
            } finally {
                // positively acknowledge a single delivery, the message will be discarded
                //channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
            }
        }, consumerTag -> {
            System.out.println("cancel" + consumerTag);
        });
        channel.basicPublish("", requestQueueName, props, message.getBytes("UTF-8"));
        String result = response.take();
        // 取消订阅队列
        channel.basicCancel(ctag);
        return result;
    }

    public void close() throws IOException {
        connection.close();
    }
}
